# @Version        : 1.0
# @Update Time    : 2025/10/15 00:00
# @File           : auth_logic.py
# @IDE            : PyCharm
# @Desc           : 认证逻辑实现类（鸭子类型）
import logging
import time
from typing import Any, Dict, Optional, Callable, Self, Sequence, List, Union, Literal

from fastapi import Response, Request

from authx.core.auth_interface import AuthInterface
from authx.core.constant import TokenConsts
from authx.core.context import AuthContextManager
from authx.core.enums import (
    ReplacedRange,
    ErrorEnum,
    LogoutRange,
    TokenErrorType,
    LogoutMode,
    LoginIdMark,
)
from authx.core.errors import (
    TokenError,
    AuthError,
    RoleException,
    PermissionException,
    InputError,
    SessionError,
)
from authx.core.event import do_logout_event, do_kickout_event, do_replaced_event
from authx.core.manager import AuthManager
from authx.core.parameter import LoginParameter, LogoutParameter
from authx.dao.base import AuthTokenDao
from authx.models.session import AuthSessionModel, TerminalInfoModel, SessionKey
from authx.models.token import TokenInfo, TokenResponse
from authx.settings.config import AuthConfig, CookieConfig
from authx.strategy.fa_strategy import AuthStrategy


logger = logging.getLogger(__name__)


class AuthLogic:
    """
    认证逻辑实现类（鸭子类型）
    """

    def __init__(
        self,
        login_type: str = "login",
        dao: Optional[AuthTokenDao] = None,
        config: Optional[AuthConfig] = None,
        auth_imp: Optional[AuthInterface] = None,
    ):
        self._login_type = login_type
        self.dao = dao or AuthManager.get_dao()
        self.config: AuthConfig = config or AuthManager.get_config()
        self.auth_imp = auth_imp or AuthManager.get_auth_interface()
        AuthManager.put_logic(self)  # type: ignore

    @property
    def login_type(self) -> str:
        """
        获取当前 StpLogic 账号类型标识
        """
        return self._login_type

    def set_login_type(self, login_type: str) -> Self:
        """
        安全的重置当前账号类型

        :param login_type: 账号类型标识
        :return: 对象自身
        """
        # 这里简化了全局管理器的实现
        # 先从全局管理器中移除旧的映射
        if self.login_type:
            AuthManager.remove_logic(self.login_type)

        # 设置新的登录类型
        self._login_type = login_type

        # 重新注册到全局管理器
        AuthManager.put_logic(self)  # type: ignore

        return self

    def get_token_name(self) -> str:
        """
        返回 token 名称
        """
        return self.config.token_name

    def create_token_value(
        self, login_id: Any, device_type: str, timeout: int, extra_data: Dict[str, Any]
    ) -> str:
        """
        为指定账号创建一个 token

        :param login_id: 账号id
        :param device_type: 设备类型
        :param timeout: 过期时间
        :param extra_data: 扩展信息
        :return: 生成的tokenValue
        """
        # 简化实现，实际应根据策略生成token
        return AuthStrategy().create_token(login_id, self.login_type)

    def set_token_value(
        self,
        response: Response,
        token_value: str,
        cookie_timeout: Optional[int] = None,
        login_parameter: Optional[LoginParameter] = None,
    ):
        """
        在当前会话写入指定 token 值
        """
        if not token_value:
            return

        # 1. 将 token 写入到当前请求的 Storage 存储器里
        self.set_token_value_to_storage(token_value)

        # 2. 将 token 写入到当前会话的 Cookie 里
        if self.config.is_read_cookie:
            cookie_cfg = login_parameter.cookie if login_parameter else None
            cookie_timeout = (
                cookie_timeout or login_parameter.timeout if login_parameter else None
            )
            self.set_token_value_to_cookie(
                response, token_value, cookie_cfg, cookie_timeout
            )

        # 3. 将 token 写入到当前请求的响应头中
        if login_parameter and login_parameter.is_write_header:
            self.set_token_value_to_response_header(response, token_value)

    def get_or_create_login_session(self, login_id: Any):
        token = self.get_token_value_by_login_id(login_id)
        return token if token else self.create_login_session(login_id)

    def set_token_value_to_storage(self, token_value: str):
        """
        将 token 写入到当前请求的 Storage 存储器里
        """
        # 简化实现
        request = self._get_request()
        storage = {}
        token_prefix = self.config.token_prefix
        request.state.token_value = token_value

        if not token_prefix:
            storage["just_created_save"] = token_value
        else:
            storage["just_created_save"] = f"{token_prefix}-{token_value}"

        # 以无前缀的方式再写入一次
        storage["just_created_not_prefix"] = token_value
        request.state.storage = storage

    def set_token_value_to_cookie(
        self,
        response: Response,
        token_value: str,
        cookie_config: Optional[CookieConfig] = None,
        cookie_timeout: Optional[int] = None,
    ):
        """
        将 token 写入到当前会话的 Cookie 里
        """
        # 简化实现
        cookie_cfg: CookieConfig = cookie_config or self.config.cookie_config

        # 实际应设置Cookie
        response.set_cookie(
            key=self.config.token_name,
            value=token_value,
            max_age=cookie_timeout,
            domain=cookie_cfg.domain,
            path=cookie_cfg.path,
            secure=cookie_cfg.secure,
            httponly=cookie_cfg.http_only,
            samesite=cookie_cfg.same_site,
        )

    def set_token_value_to_response_header(self, response: Response, token_value: str):
        """
        将 token 写入到当前请求的响应头中
        """
        token_name = self.config.token_name
        response.headers[token_name] = token_value
        response.headers["Access-Control-Allow-Origin"] = token_name

    def get_token_value(self, no_prefix_throw_exception=False) -> Optional[str]:
        """
        获取当前请求的 token 值
        """
        # 1. 获取前端提交的 token (包含前缀值)
        return AuthContextManager.get_current_token()

    def get_token_value_not_cut(self) -> Optional[str]:
        """
        获取当前请求的 token 值 (不裁剪前缀)
        """
        # 使用缓存的请求对象
        request = self._get_cached_request()
        storage = getattr(request.state, "storage", {})
        key_token_name = self.config.token_name
        token_value = None

        # 1. 先尝试从 Storage 存储器里读取
        if "just_created_save" in storage:
            token_value = storage["just_created_save"]

        # 2. 再尝试从 header 头里读取
        if not token_value and self.config.is_read_header:
            token_value = request.headers.get(key_token_name)

        # 3. 最后尝试从 cookie 里读取
        if not token_value and self.config.is_read_cookie:
            token_value = request.headers.get(key_token_name)
            if token_value and self.config.cookie_auto_fill_prefix:
                token_value = f"{self.config.token_prefix}-{token_value}"

        return token_value

    def get_token_value_not_null(self) -> str:
        """
        获取当前请求的 token 值，如果获取不到则抛出异常
        """
        # 使用缓存的请求对象提高性能
        token_value = self.get_token_value(True)
        if not token_value:
            raise TokenError(
                error_type=TokenErrorType.NOT_TOKEN,
                context=dict(
                    login_type=self._login_type,
                ),
            )
        return token_value

    async def get_token_info(self) -> TokenInfo:
        """
        获取当前会话的 token 参数信息
        """
        # 简化实现
        return TokenInfo.model_validate(
            {
                "token_name": self.config.token_name,
                "token_value": self.get_token_value(),
                "is_login": self.is_login(),
                "login_id": await self.get_login_id_default_null(),
                "login_type": self._login_type,
                "token_timeout": await self.get_token_timeout(),
                "session_timeout": await self.get_session_timeout(),
                "token_session_timeout": await self.get_token_session_timeout(),
                "token_active_timeout": await self.get_token_active_timeout(),
                "login_device_type": await self.get_login_device_type(),
            }
        )

    async def login(
        self,
        *,
        response: Response,
        login_id: Any,
        device_type: Optional[str] = None,
        is_lasting_cookie: Optional[bool] = None,
        timeout: Optional[int] = None,
        login_parameter: Optional[LoginParameter] = None,
    ):
        """

        :param response:
        :param login_id:
        :param device_type:
        :param is_lasting_cookie:
        :param timeout:
        :param login_parameter:
        :return:
        """
        if login_parameter is None:
            login_parameter = self.create_sa_login_parameter()

        if device_type is not None:
            login_parameter.device_type = device_type

        if is_lasting_cookie is not None:
            login_parameter.is_lasting_cookie = is_lasting_cookie

        if timeout is not None:
            login_parameter.timeout = timeout

        # 1. 创建会话
        token = await self.create_login_session(login_id, login_parameter)
        request = self._get_request()
        request.state.token = token

        # 2. 在当前客户端注入 token
        self.set_token_value(response, token, login_parameter=login_parameter)

    def result_ok(self) -> TokenResponse:
        request = self._get_request()
        token = getattr(request.state, "token", "")
        return TokenResponse.model_validate(
            {"access_token": token, "token_type": "bearer"}
        )

    async def create_login_session(
        self, login_id: Any, login_parameter: LoginParameter = None
    ) -> str:
        """
        创建指定账号 id 的登录会话数据
        """
        if login_parameter is None:
            login_parameter = self.create_sa_login_parameter()

        # 1. 先检查一下，传入的参数是否有效
        self._check_login_args(login_id, login_parameter)

        # 2. 给这个账号分配一个可用的 token
        token_value = await self._dist_usable_token(login_id, login_parameter)

        # 3. 获取此账号的 Account-Session, 续期
        session = await self.get_session_by_login_id(
            login_id, True, login_parameter.timeout
        )

        await session.update_min_timeout(login_parameter.timeout)

        # 4. 在 Account-Session 上记录本次登录的终端信息
        terminal_info = TerminalInfoModel(
            **{
                "device_type": login_parameter.device_type,
                "device_id": login_parameter.device_id,
                "token_value": token_value,
                "extra_data": login_parameter.extra_data,
            }
        )
        await session.add_terminal(terminal_info)

        # 5. 保存 token -> id 的映射关系
        await self.save_token_to_id_mapping(
            token_value, login_id, login_parameter.timeout
        )

        # 6. 写入这个 token 的最后活跃时间
        if self._is_open_check_active_timeout():
            await self._set_last_active_to_now(
                token_value, login_parameter.active_timeout, login_parameter.timeout
            )

        # 7. 如果该 token 对应的 Token-Session 已经存在，则需要给其续期
        token_session = await self.get_token_session_by_token(
            token_value, login_parameter.right_now_create_token_session
        )
        if token_session:
            await token_session.update_min_timeout(login_parameter.timeout)

        # 8. 检查此账号会话数量是否超出最大值
        if login_parameter.max_login_count != -1:
            await self.logout_by_max_login_count(
                login_id,
                session,
                None,
                login_parameter.max_login_count,
                login_parameter.overflow_logout_mode,
            )

        return token_value

    async def _dist_usable_token(
        self, login_id: Any, login_parameter: LoginParameter
    ) -> str:
        """
        为指定账号 id 的登录操作，分配一个可用的 token
        """
        # 1. 如果不允许一个账号多地同时登录，则需要先将这个账号的历史登录会话标记为被顶下线
        if not login_parameter.is_concurrent:
            if login_parameter.replaced_range == ReplacedRange.CURR_DEVICE_TYPE:
                await self.replaced(login_id, login_parameter.device_type)
            elif login_parameter.replaced_range == ReplacedRange.ALL_DEVICE_TYPE:
                await self.replaced(login_id, None, self.create_logout_parameter())

        # 2. 如果调用者预定了要生成的 token，则直接返回这个预定的值
        if login_parameter.token:
            return login_parameter.token

        # 3.只有在配置了 [ 允许一个账号多地同时登录 ] 时，才尝试复用旧 token，这样可以避免不必要地查询，节省开销
        if login_parameter.is_concurrent:
            if self.is_support_share_token() and login_parameter.is_share:
                token_value = await self.get_token_value_by_login_id(
                    login_id, login_parameter.device_type
                )
                if token_value:
                    return token_value

        # 4. 新建 token
        # 获取FaStrategy单例实例，而不是每次创建新实例
        strategy = AuthStrategy()

        try:
            return await strategy.generate_unique_token(
                "token",
                self._get_config_of_max_try_times(login_parameter),
                # 更清晰的命名和参数传递
                self._create_token_generator(login_id, login_parameter),
                self._check_token_function,
            )
        except AuthError as e:
            # 增强异常处理，添加更多上下文信息
            raise AuthError(
                "token_generation_failed",
                f"生成登录token失败：{str(e)}",
                context={
                    "login_type": self._login_type,
                    "login_id": login_id,
                    "device_type": login_parameter.device_type,
                },
                cause=e,
            ) from e

    def _create_token_generator(
        self, login_id: Any, login_parameter: LoginParameter
    ) -> Callable[[], str]:
        """
        创建token生成器函数

        :param login_id: 登录ID
        :param login_parameter: 登录参数
        :return: token生成器函数
        """

        def generate_token() -> str:
            # 使用配置的超时时间，如果login_parameter.timeout为None则使用默认值
            timeout = (
                login_parameter.timeout
                if login_parameter.timeout is not None
                else self.config.timeout
            )
            return self.create_token_value(
                login_id,
                login_parameter.device_type,
                timeout,
                login_parameter.extra_data,
            )

        return generate_token

    async def _check_token_function(self, token_value: str):
        login_id = await self.get_login_id_not_handle(token_value)
        return login_id is not None

    def _check_login_args(self, login_id: Any, login_parameter: LoginParameter):
        """
        校验登录时的参数有效性
        """
        if not login_id:
            raise InputError(
                "loginId 不能为空",
                context={"login_type": self._login_type},
            )

        # 检查id是否为异常标记值
        if str(login_id) in ErrorEnum.get_abnormal_list():
            raise InputError(
                f"loginId 不能为以下值：{ErrorEnum.get_abnormal_list()}",
                context={"login_type": self._login_type, "login_id": str(login_id)},
            )

        # 判断当前 StpLogic 是否支持 extra 扩展参数
        if not self.is_support_extra() and login_parameter.extra_data:
            logger.warning(
                "警告：当前 StpLogic 不支持 extra 扩展参数模式，传入的 extra 参数将被忽略"
            )

        # 如果全局配置未启动动态 activeTimeout 功能，但是此次登录却传入了 activeTimeout 参数
        if (
            not self.config.dynamic_active_timeout
            and login_parameter.active_timeout is not None
        ):
            logger.warning(
                "警告：当前全局配置未开启动态 activeTimeout 功能，传入的 activeTimeout 参数将被忽略"
            )

    async def logout(
        self,
        response: Response,
        *,
        login_id: Optional[Any] = None,
        device_type: Optional[str] = None,
        logout_parameter: Optional[LogoutParameter] = None,
    ):
        """
        会话注销方法，支持多种参数形式

        Args:
            response: 响应对象
            login_id: 账号ID (可选)
            device_type: 设备类型 (可选，填None代表注销该账号的所有设备类型)
            logout_parameter: 注销参数对象 (可选)
        """
        logout_parameter = logout_parameter or self.create_logout_parameter()
        if login_id is None:
            if device_type is not None:
                logout_parameter.device_type = device_type
            token_value = self.get_token_value()
            if not token_value:
                return
            if self.config.is_read_cookie:
                response.delete_cookie(self.config.token_name)
            if logout_parameter.range == LogoutRange.TOKEN:
                await self.logout_by_token_value(token_value)
            else:
                login_id = (
                    login_id
                    or await self.get_login_id_by_token_not_think_freeze(token_value)
                )
                if login_id:
                    if logout_parameter.is_keep_freeze_ops and self.is_freeze(
                        token_value
                    ):
                        return
                    await self._logout(
                        login_id=login_id, logout_parameter=logout_parameter
                    )
        else:
            await self._logout(login_id=login_id, logout_parameter=logout_parameter)

    async def logout_by_token_value(
        self, token_value: str, logout_parameter: LogoutParameter = None
    ):
        """
        注销下线，根据指定 token
        """
        if logout_parameter is None:
            logout_parameter = self.create_sa_logout_parameter()
        await self._logout_by_token_value(token_value, logout_parameter)

    async def _publish_logout_event(self, event_type: str, login_id, token_value):
        """
        发布登出相关事件

        Args:
            event_type: 事件类型 ("LOGOUT", "KICKOUT", "REPLACED")
            login_id: 登录ID
            token_value: token值
        """
        event_map = {
            "LOGOUT": (do_logout_event, "do_logout_event"),
            "KICKOUT": (do_kickout_event, "do_kickout_event"),
            "REPLACED": (do_replaced_event, "do_replaced_event"),
        }

        if event_type in event_map:
            event, event_name = event_map[event_type]
            # 直接使用同步版本的send方法，因为所有事件处理函数都是同步的
            event.send(
                event_name,
                login_type=self.login_type,
                login_id=login_id,
                token_value=token_value,
            )

    async def _logout_by_token_value(
        self, token_value: str, logout_parameter: LogoutParameter
    ):
        """
        注销下线，根据指定 token、注销参数
        """
        # 1. 判断一下：如果此 token 映射的是一个无效 loginId，则立即返回
        login_id = await self.get_login_id_by_token_not_think_freeze(token_value)
        if not login_id:
            return
        if not logout_parameter.is_keep_freeze_ops and await self.is_freeze(
            token_value
        ):
            return

        # 2. 清除这个 token 的最后活跃时间记录
        if self._is_open_check_active_timeout():
            await self._clear_last_active(token_value)

        # 3. 清除 Token-Session
        if not logout_parameter.is_keep_token_session:
            await self.delete_token_session(token_value)

        # 4. 清理或更改 Token 映射
        # 5. 发布事件通知
        mode = logout_parameter.mode.value
        if mode == LogoutMode.LOGOUT:
            await self.delete_token_to_id_mapping(token_value)
            # 使用统一的事件发布方法
            await self._publish_logout_event(mode, login_id, token_value)
        elif mode == LogoutMode.KICKOUT:
            await self.update_token_to_id_mapping(token_value, LoginIdMark.KICK_OUT)
            # 使用统一的事件发布方法
            await self._publish_logout_event(mode, login_id, token_value)
        elif mode == LogoutMode.REPLACED:
            await self.update_token_to_id_mapping(token_value, LoginIdMark.BE_REPLACED)
            # 使用统一的事件发布方法
            await self._publish_logout_event(mode, login_id, token_value)
        # 6. 清理这个账号的 Account-Session 上的 terminal 信息
        session = await self.get_session_by_login_id(login_id, False)
        if session:
            await session.remove_terminal(token_value)
            await session.logout_by_terminal_count_to_zero()

    async def is_login(self, login_id: Optional[Any] = None) -> bool:
        """
        判断当前会话是否已经登录
        """
        if login_id is None:
            return bool(await self.get_login_id_default_null())
        terminal_list = await self.get_terminal_list_by_login_id(login_id)
        return bool(terminal_list)

    async def check_login(self):
        """
        检验当前会话是否已经登录，如未登录，则抛出异常
        """
        await self.get_login_id()

    async def get_login_id(self) -> Any:
        """
        获取当前会话账号id，如果未登录，则抛出异常
        """
        # 1. 先判断一下当前会话是否正在临时身份切换
        if self.is_switch():
            return self.get_switch_login_id()

        # 2. 如果前端没有提交 token，则抛出异常
        token_value = self.get_token_value(True)
        if not token_value:
            raise TokenError(
                error_type=TokenErrorType.NOT_TOKEN,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

        # 3. 查找此 token 对应的 loginId，如果找不到则抛出异常
        login_id = await self.get_login_id_not_handle(token_value)
        if not login_id:
            raise TokenError(
                error_type=TokenErrorType.INVALID_TOKEN,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

        # 4. 如果这个 token 指向的是过期标记，则抛出异常
        if login_id == "TOKEN_TIMEOUT":
            raise TokenError(
                error_type=TokenErrorType.TOKEN_TIMEOUT,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

        # 5. 如果这个 token 指向的是被顶替标记，则抛出异常
        if login_id == LoginIdMark.BE_REPLACED:
            raise TokenError(
                error_type=TokenErrorType.BE_REPLACED,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

        # 6. 如果这个 token 指向的是被踢下线标记，则抛出异常
        if login_id == LoginIdMark.KICK_OUT:
            raise TokenError(
                error_type=TokenErrorType.KICK_OUT,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

        # 7. token 活跃频率检查
        await self.check_active_timeout_by_config(token_value)

        return login_id

    async def get_login_id_default_null(self) -> Any:
        """
        获取当前会话账号id, 如果未登录，则返回null
        """
        # 1. 先判断一下当前会话是否正在临时身份切换
        if self.is_switch():
            return self.get_switch_login_id()

        # 2. 如果前端连 token 都没有提交，则直接返回 null
        token_value = self.get_token_value()
        if not token_value:
            return None

        # 3. 根据 token 找到对应的 loginId
        login_id = await self.get_login_id_not_handle(token_value)
        if not self._is_valid_login_id(login_id):
            return None

        # 4. 如果 token 已被冻结，也返回 null
        if (
            await self.get_token_active_timeout_by_token(token_value)
            == AuthTokenDao.NOT_VALUE_EXPIRE
        ):
            return None

        return login_id

    async def get_login_id_by_token(self, token_value: str) -> Any:
        """
        获取指定 token 对应的账号id
        """
        login_id = await self.get_login_id_by_token_not_think_freeze(token_value)
        if (
            login_id and await self.get_token_active_timeout_by_token(token_value) != -2
        ):  # NOT_VALUE_EXPIRE
            return login_id
        return None

    async def get_login_id_by_token_not_think_freeze(self, token_value: str) -> Any:
        """
        获取指定 token 对应的账号id (不考虑被冻结)
        """
        if not token_value:
            return None

        login_id = await self.get_login_id_not_handle(token_value)
        if not self._is_valid_login_id(login_id):
            return None

        return login_id

    async def get_login_id_not_handle(self, token_value: str) -> Any:
        """
        获取指定 token 对应的账号id (不做任何特殊处理)
        """
        # 简化实现，实际应从持久化层获取
        return await self.dao.get(self._splicing_key_token_value(token_value))

    def _is_valid_login_id(self, login_id: Any) -> bool:
        """
        判断一个 loginId 是否是有效的
        """
        if not login_id:
            return False

        abnormal_list = ErrorEnum.get_abnormal_list()
        return str(login_id) not in abnormal_list

    async def save_token_to_id_mapping(
        self, token_value: str, login_id: Any, timeout: int
    ):
        """
        存储 token - id 映射关系
        """
        await self.dao.set(
            self._splicing_key_token_value(token_value), str(login_id), timeout
        )

    async def update_token_to_id_mapping(self, token_value: str, login_id: Any):
        """
        更改 token - id 映射关系
        """
        if not login_id:
            raise InputError(
                "loginId 不能为空",
                context={"login_type": self._login_type},
            )
        await self.dao.update(
            self._splicing_key_token_value(token_value), str(login_id)
        )

    async def delete_token_to_id_mapping(self, token_value: str):
        """
        删除 token - id 映射
        """
        await self.dao.delete(self._splicing_key_token_value(token_value))

    async def _set_last_active_to_now(
        self,
        token_value: str,
        active_timeout: Optional[int] = None,
        timeout: Optional[int] = None,
    ):
        """
        写入指定 token 的最后活跃时间为当前时间戳
        """
        if timeout is None:
            timeout = self.config.timeout

        # 将此 token 的最后活跃时间标记为当前时间戳
        key = self._splicing_key_last_active_time(token_value)
        value = str(int(time.time() * 1000))
        if self.config.dynamic_active_timeout and active_timeout is not None:
            value += f",{active_timeout}"
        await self.dao.set(key, value, timeout)

    async def _clear_last_active(self, token_value: str):
        """
        清除指定 Token 的最后活跃时间记录
        """
        await self.dao.delete(self._splicing_key_last_active_time(token_value))

    async def clear_last_active(self, token_value: str):
        """
        清除指定 Token 的最后活跃时间记录
        """
        await self.dao.delete(self._splicing_key_last_active_time(token_value))

    async def is_freeze(self, token_value: str) -> bool:
        """
        判断指定 token 是否已被冻结
        """
        # 获取这个 token 的剩余活跃有效期
        active_timeout = await self.get_token_active_timeout_by_token(token_value)

        # 值为 -1 代表此 token 已经被设置永不冻结
        if active_timeout == AuthTokenDao.NEVER_EXPIRE:  # NEVER_EXPIRE
            return False

        # 值为 -2 代表已被冻结
        if active_timeout == AuthTokenDao.NOT_VALUE_EXPIRE:  # NOT_VALUE_EXPIRE
            return True

        return False

    async def check_active_timeout_by_config(self, token_value: str):
        """
        根据全局配置决定是否校验指定 token 的活跃度
        """
        if self._is_open_check_active_timeout():
            storage = self._get_storage()
            request = self._get_request()
            # 可以避免一次请求多次校验
            if "token_active_timeout_checked_key" not in storage:
                storage["token_active_timeout_checked_key"] = True
                request.state.storage = storage

                # 1. 检查此 token 的最后活跃时间是否已经超过了 active-timeout 的限制
                self.check_active_timeout(token_value)

                # 2. 如果配置了自动续签功能，则更新这个 token 的最后活跃时间
                if self._should_auto_renew():
                    await self.update_last_active_to_now(token_value)

    async def get_token_use_active_timeout_or_global_config(
        self, token_value: str
    ) -> int:
        """
        获取指定 token 在缓存中的 activeTimeout 值，如果不存在则返回全局配置的 activeTimeout 值
        """
        active_timeout = await self.get_token_use_active_timeout(token_value)
        if active_timeout is None:
            return self.config.active_timeout
        return active_timeout

    async def get_token_last_active_time(
        self, token_value: Optional[str] = None
    ) -> int:
        """
        获取指定 token 的最后活跃时间（13位时间戳）
        """
        if token_value is None:
            token_value = self.get_token_value()
        if not token_value:
            return -2  # NOT_VALUE_EXPIRE

        # 获取这个 token 的最后活跃时间，13位时间戳
        key = self._splicing_key_last_active_time(token_value)
        last_active_time_string = await self.dao.get(key)

        # 查不到，返回-2
        if not last_active_time_string:
            return -2  # NOT_VALUE_EXPIRE

        # 根据逗号切割字符串
        if "," in last_active_time_string:
            return int(last_active_time_string.split(",")[0])  # noqa
        return int(last_active_time_string)

    async def get_token_timeout(self, token: Optional[str] = None) -> int:
        """
        获取 token 剩余有效时间
        """
        if token is None:
            token = self.get_token_value()
        return await self.dao.get_timeout(self._splicing_key_token_value(token))

    async def get_session_timeout_by_login_id(self, login_id: Any) -> int:
        """
        获取指定账号 id 的 Account-Session 剩余有效时间
        """
        return await self.dao.get_session_timeout(self._splicing_key_session(login_id))

    async def get_token_session_timeout_by_token_value(self, token_value: str) -> int:
        """
        获取指定 token 的 Token-Session 剩余有效时间
        """
        return await self.dao.get_session_timeout(
            self._splicing_key_token_session(token_value)
        )

    async def get_token_active_timeout_by_token(
        self, token_value: Optional[str] = None
    ) -> int:
        """
        获取指定 token 剩余活跃有效期
        """
        if token_value is None:
            token_value = self.get_token_value()

        # 如果全局配置了永不冻结，则返回 -1
        if not self._is_open_check_active_timeout():
            return -1  # NEVER_EXPIRE

        # 先获取这个 token 的最后活跃时间，13位时间戳
        last_active_time = await self.get_token_last_active_time(token_value)
        if last_active_time == -2:  # NOT_VALUE_EXPIRE
            return -2  # NOT_VALUE_EXPIRE

        # 实际时间差
        time_diff = (int(time.time() * 1000) - last_active_time) // 1000
        # 该 token 允许的时间差
        allow_time_diff = await self.get_token_use_active_timeout_or_global_config(
            token_value
        )
        if allow_time_diff == -1:  # NEVER_EXPIRE
            return -1  # NEVER_EXPIRE

        # 校验这个时间差是否超过了允许的值
        active_timeout = allow_time_diff - time_diff
        if active_timeout < 0:
            return -2  # NOT_VALUE_EXPIRE
        else:
            return active_timeout

    async def renew_timeout(self, timeout: int, token_value: Optional[str] = None):
        """
        对 token 的 timeout 值进行续期
        """
        if token_value is None:
            token_value = self.get_token_value()

        # 1. 如果 token 指向的 loginId 为空，或者属于异常项时，不进行续期操作
        login_id = await self.get_login_id_by_token(token_value)
        if not login_id:
            return

        # 2. 检查 token 合法性
        session = await self.get_session_by_login_id(login_id)
        if not session:
            raise SessionError(
                message=f"未能查询到对应 Access-Session 会话，无法续期 login_type={self._login_type}",
            )
        if not session.get_terminal(token_value):
            raise SessionError(
                message=f"未能查询到对应终端信息，无法续期 login_type={self._login_type}",
            )

        # 3. 续期此 token 本身的有效期
        await self.dao.update_timeout(
            self._splicing_key_token_value(token_value), timeout
        )

        # 4. 续期此 token 的 Token-Session 有效期
        token_session = await self.get_token_session_by_token(token_value, False)
        if token_session:
            await token_session.update_timeout(timeout)

        # 5. 续期此 token 指向的账号的 Account-Session 有效期
        await session.update_min_timeout(timeout)

        # 6. 更新此 token 的最后活跃时间
        if self._is_open_check_active_timeout():
            await self.dao.update_timeout(
                self._splicing_key_last_active_time(token_value), timeout
            )

    async def get_role_list(self, login_id: Optional[Any] = None) -> Sequence[str]:
        """
        获取账号的角色集合
        """
        if login_id is None:
            login_id = await self.get_login_id()
        # 从缓存获取
        session = await self.get_session_by_login_id(login_id)
        cache_roles = session.get(SessionKey.ROLE_LIST.value)
        if cache_roles is not None:
            return cache_roles
        # 简化实现，实际应从权限接口获取
        roles = await AuthManager.get_auth_interface().get_role_list(
            login_id, self.login_type
        )
        if cache_roles is None:
            await session.set(SessionKey.ROLE_LIST.value, roles)
        return roles

    async def has_single_role(self, role: str, login_id: Optional[Any] = None) -> bool:
        """
        判断账号是否拥有指定角色
        """
        if login_id is None:
            try:
                login_id = await self.get_login_id()
            except Exception:
                return False
        return role in await self.get_role_list(login_id)

    def check_single_role(self, role: str):
        """
        校验当前账号是否含有指定角色标识
        """
        if not self.has_single_role(role, self.get_login_id()):
            raise RoleException(login_type=self._login_type, required=[role])

    async def get_permission_list(
        self, login_id: Optional[Any] = None
    ) -> Sequence[str]:
        """
        获取账号的权限码集合
        """
        if login_id is None:
            login_id = await self.get_login_id()
        # 从缓存获取
        session = await self.get_session_by_login_id(login_id)
        cache_permission_list = session.get(SessionKey.PERMISSION_LIST.value)
        if cache_permission_list is not None:
            return cache_permission_list
        # 简化实现，实际应从权限接口获取
        permission_list = await AuthManager.get_auth_interface().get_permission_list(
            login_id, self._login_type
        )
        # 存入缓存，只在权限列表不为空时缓存
        if permission_list is not None:
            await session.set(SessionKey.PERMISSION_LIST.value, permission_list)
        return permission_list

    async def is_super_admin(self, login_id: Optional[Any] = None) -> bool:
        """
        判断当前账号是否是超级管理员
        """
        if login_id is None:
            login_id = await self.get_login_id()
        # 简化实现，实际应从权限接口获取
        return await AuthManager.get_auth_interface().is_super_admin(
            login_id, self._login_type
        )

    async def has_single_permission(
        self, permission: str, login_id: Optional[Any] = None
    ) -> bool:
        """
        判断账号是否含有指定权限
        """
        if login_id is None:
            try:
                login_id = self.get_login_id()
            except Exception as e:  # noqa
                return False
        return permission in await self.get_permission_list(login_id)

    def check_single_permission(self, permission: str):
        """
        校验当前账号是否含有指定权限
        """
        if not self.has_single_permission(permission, self.get_login_id()):
            raise PermissionException(
                login_type=self._login_type,
                required=[permission],
            )

    async def get_token_value_by_login_id(
        self, login_id: Any, device_type: Optional[str] = None
    ) -> Optional[str]:
        """
        获取指定账号 id 的 token
        """
        token_value_list = await self.get_token_value_list_by_login_id(
            login_id, device_type
        )

        return token_value_list[-1] if token_value_list else None  # noqa

    async def get_token_value_list_by_login_id(
        self, login_id: Any, device_type: Optional[str] = None
    ) -> List[str]:
        """
        获取指定账号 id 的 token 集合
        """
        # 如果该账号的 Account-Session 为 null，说明此账号尚没有客户端在登录
        session = await self.get_session_by_login_id(login_id, False)
        if not session:
            return []

        # 按照设备类型进行筛选
        return session.get_token_value_list_by_device_type(device_type)

    async def get_login_device_type(
        self, token_value: Optional[str] = None
    ) -> Optional[str]:
        """
        返回会话的登录设备类型
        """
        if token_value is None:
            token_value = self.get_token_value()
        terminal_info = await self.get_terminal_info_by_token(token_value)
        return terminal_info.device_type if terminal_info else None

    async def get_terminal_info_by_token(
        self, token_value: str
    ) -> Optional[TerminalInfoModel]:
        """
        返回指定 token 指向的设备信息
        """
        if not token_value:
            return None

        # 判断 Token 是否有效
        login_id = self.get_login_id_not_handle(token_value)
        if not self._is_valid_login_id(login_id):
            return None

        # 判断 Account-Session 是否存在
        session = await self.get_session_by_login_id(login_id, False)
        if not session:
            return None

        # 判断 Token 是否已被冻结
        if self.is_freeze(token_value):
            return None

        # 遍历 Account-Session 上的客户端 token 列表
        terminal_list = session.terminal_list_copy()
        for terminal in terminal_list:
            if terminal.token_value == token_value:
                return terminal

        return None

    def switch_to(self, login_id: Any):
        """
        临时切换身份为指定账号id
        """
        storage = self._get_storage()
        storage[self._splicing_key_switch()] = login_id

    def end_switch(self):
        """
        结束临时切换身份
        """
        storage = self._get_storage()
        if self._splicing_key_switch() in storage:
            del storage[self._splicing_key_switch()]

    def is_switch(self) -> bool:
        """
        判断当前请求是否正处于身份临时切换中
        """
        storage = self._get_storage()
        return self._splicing_key_switch() in storage

    def get_switch_login_id(self) -> Any:
        """
        返回身份临时切换的 loginId
        """
        storage = self._get_storage()
        return storage.get(self._splicing_key_switch())

    async def open_safe(self, safe_time: int, service: str = "default"):
        """
        在当前会话开启二级认证
        """
        # 1. 开启二级认证前必须处于登录状态
        await self.check_login()

        # 2. 写入指定的可以标记，打开二级认证
        token_value = self.get_token_value_not_null()
        await self.dao.set(
            self._splicing_key_safe(token_value, service), "safe", safe_time
        )

    def is_safe(self, service: str = "default") -> bool:
        """
        判断当前会话是否处于二级认证时间内
        """
        return self.is_safe_by_token(self.get_token_value(), service)

    def is_safe_by_token(self, token_value: str, service: str) -> bool:
        """
        判断指定 token 是否处于二级认证时间内
        """
        if not token_value:
            return False

        # 如果此 token 不处于登录状态，也将其视为未认证
        login_id = self.get_login_id_not_handle(token_value)
        if not self._is_valid_login_id(login_id):
            return False

        # 如果缓存中可以查询出指定的键值，则代表已认证
        value = self.dao.get(self._splicing_key_safe(token_value, service))
        return bool(value)

    def check_safe(self, service: str = "default"):
        """
        校验当前会话是否已通过二级认证
        """
        # 1. 必须先通过登录校验
        self.check_login()

        # 2. 再进行二级认证校验
        token_value = self.get_token_value()
        value = self.dao.get(self._splicing_key_safe(token_value, service))
        if not value:
            raise Exception("未通过二级认证")

    async def get_safe_time(self, service: str = "default") -> int:
        """
        获取当前会话的二级认证剩余有效时间
        """
        token_value = self.get_token_value()
        if not token_value:
            return -2  # NOT_VALUE_EXPIRE

        # 从缓存中查询这个 key 的剩余有效期
        return await self.dao.get_timeout(self._splicing_key_safe(token_value, service))

    async def close_safe(self, service: str = "default"):
        """
        在当前会话结束二级认证
        """
        token_value = self.get_token_value()
        if not token_value:
            return

        # 删除 key
        await self.dao.delete(self._splicing_key_safe(token_value, service))

    # ------------------- 拼接相应key -------------------

    def _splicing_key_token_value(self, token_value: str) -> str:
        """
        拼接保存 token - id 映射关系时使用的key
        """
        return f"{self.config.token_name}:{self._login_type}:token:{token_value}"

    def _splicing_key_session(self, login_id: Any) -> str:
        """
        拼接保存 Account-Session 时使用的 key
        """
        return f"{self.config.token_name}:{self._login_type}:session:{login_id}"

    def _splicing_key_token_session(self, token_value: str) -> str:
        """
        拼接保存 Token-Session 时使用的 key
        """
        return (
            f"{self.config.token_name}:{self._login_type}:token-session:{token_value}"
        )

    def _splicing_key_last_active_time(self, token_value: str) -> str:
        """
        拼接保存 token 最后活跃时间时使用的 key
        """
        return f"{self.config.token_name}:{self._login_type}:last-active:{token_value}"

    def _splicing_key_switch(self) -> str:
        """
        拼接进行临时身份切换时使用的 key
        """
        return f"switch_to_save_key_{self._login_type}"

    def _splicing_key_safe(self, token_value: str, service: str) -> str:
        """
        拼接保存业务二级认证标记时使用的 key
        """
        return (
            f"{self.config.token_name}:{self._login_type}:safe:{service}:{token_value}"
        )

    def _splicing_key_disable(self, login_id: Any, service: Optional[str] = None):
        """
        拼接保存账号封禁标记时使用的 key
        """
        service = service or TokenConsts.DEFAULT_DISABLE_SERVICE
        return (
            f"{self.config.token_name}:{self._login_type}:disable:{service}:{login_id}"
        )

    # ------------------- 辅助方法 -------------------

    # 权限匹配支持 and /or 逻辑 支持通配符 *
    def has_element(
        self,
        source_elements: Sequence[str],
        target_patterns: Union[str, Sequence[str]],
        logic: Literal["and", "or"] = "and",
    ):
        """
        检查元素序列是否匹配字符串，支持通配符 * 模式匹配（仅允许在末尾使用）

        通配符匹配规则：
        - 通配符 * 只能出现在模式末尾
        - 无通配符时，进行精确匹配
        - 通配符 * 在末尾时，匹配以通配符前的部分开头的权限
        - 例如：'admin:*' 可以匹配 'admin:user:add', 'admin:view' 等
        - 不允许：'*:delete', 'api:*:read' 等格式的通配符模式

        Args:
            source_elements: 用户的权限序列（会自动去重处理）
            target_patterns: 需要校验的权限（单个权限或权限序列，序列会自动去重处理）
            logic: 逻辑运算符，当输入为序列时有效（"and"表示需要满足所有条件，"or"表示满足任一条件）

        Returns:
            bool: 是否匹配成功

        Raises:
            ValueError: 如果通配符使用不当（非末尾位置或多个通配符）
        """

        # 处理单个权限的情况
        def _match_single(element_str: str, match_pattern: str) -> bool:
            """检查单个权限是否匹配通配符模式

            通配符规则：
            - 通配符 * 只能出现在模式末尾
            - 无通配符时，进行精确匹配
            - 通配符 * 在末尾时，匹配以通配符前的部分开头的权限
            """
            # 如果没有通配符，直接比较
            if "*" not in match_pattern:
                return element_str == match_pattern

            # 确保通配符只在末尾
            if match_pattern.count("*") > 1 or not match_pattern.endswith("*"):
                raise ValueError(f"通配符 '*' 只能出现在模式末尾: {match_pattern}")

            # 移除末尾的通配符，检查元素是否以此前缀开头
            prefix = match_pattern[:-1]  # 移除末尾的 *
            return element_str.startswith(prefix)

        # 统一转换为列表处理
        source_elements_list = list(set(source_elements))  # 去重处理
        target_patterns_list = (
            [target_patterns]
            if isinstance(target_patterns, str)
            else list(set(target_patterns))
        )  # 去重处理

        # 确保输入是列表
        if not isinstance(source_elements_list, list):
            raise TypeError("source_elements must be a sequence of strings")
        if not isinstance(target_patterns_list, list):
            raise TypeError("target_patterns must be a string or sequence of strings")

        # 当target_patterns_list为空时，默认没有权限要求
        if not target_patterns_list:
            return True

        # 当source_elements_list为空时，无法满足任何权限要求
        if not source_elements_list:
            return False

        # 根据逻辑运算符处理
        if logic == "and":
            # AND逻辑：对于每个需要匹配的模式，必须至少有一个用户权限与之匹配
            for pattern in target_patterns_list:
                if not any(_match_single(ele, pattern) for ele in source_elements_list):
                    return False
            return True
        else:  # logic == "or"
            # OR逻辑：对于所有需要匹配的模式，只要有一个模式能被至少一个用户权限匹配
            for pattern in target_patterns_list:
                if any(_match_single(ele, pattern) for ele in source_elements_list):
                    return True
            return False

    # def _get_cached_token_dao(self) -> AuthTokenDao:
    #     """
    #     获取缓存的DAO对象，如果不存在则创建
    #
    #     Returns:
    #         AuthTokenDao: DAO对象
    #     """
    #     if self._cached_token_dao is None:
    #         self._cached_token_dao = self._get_token_dao()
    #     return self._cached_token_dao

    # def _get_token_dao(self) -> AuthTokenDao:
    #     """
    #     返回当前 StpLogic 使用的持久化对象 (简化实现)
    #     """
    #     # 这里应该返回全局的DAO对象，简化实现
    #     return get_stp_logic_dao()

    def _get_storage(self):
        """
        获取当前请求的 Storage 存储器 (简化实现)
        """
        # 这里应该返回当前请求的Storage对象，简化实现
        request = self._get_request()
        return getattr(request.state, "storage", {})

    def _get_cached_request(self) -> Request:
        """
        获取缓存的请求对象，如果不存在则创建

        Returns:
            Request: 请求对象
        """
        self._cached_request = self._get_request()
        return self._cached_request

    def _get_request(self) -> Request:
        """
        获取当前请求对象 (简化实现)
        """
        # 这里应该返回当前请求对象，简化实现
        return AuthContextManager.get_request()

    def is_support_share_token(self) -> bool:
        """
        返回当前 StpLogic 是否支持共享 token 策略
        """
        return self.config.is_share

    def _is_open_check_active_timeout(self) -> bool:
        """
        返回全局配置是否开启了 Token 活跃度校验
        """
        return self.config.active_timeout != -1 or self.config.dynamic_active_timeout

    def _get_config_of_max_try_times(self, login_parameter: LoginParameter) -> int:
        """
        返回全局配置的 maxTryTimes 值
        """
        return login_parameter.max_try_times

    def _generate_unique_token(
        self,
        token_type: str,
        max_try_times: int,
        create_func: Callable,
        check_func: Callable,
    ) -> str:
        """
        生成唯一 token
        """
        # 简化实现
        return create_func()

    def _should_auto_renew(self) -> bool:
        """
        判断是否应该自动续签
        """
        # 简化实现
        return True

    def _create_session(self, session_id: str) -> AuthSessionModel:
        """
        创建 Session 对象 (简化实现)
        """
        # 这里应该返回Session对象，简化实现
        return AuthStrategy().create_session(session_id)

    def is_support_extra(self) -> bool:
        """
        当前 StpLogic 对象是否支持 token 扩展参数
        """
        return False

    def create_sa_login_parameter(self) -> LoginParameter:
        """
        创建 LoginParameter 对象
        """
        login_parameter = LoginParameter.from_config(self.config)
        return login_parameter

    def create_sa_logout_parameter(self) -> LogoutParameter:
        """
        创建 LogoutParameter 对象
        """
        return LogoutParameter.from_config(self.config)

    def create_logout_parameter(self):
        """
        创建 LogoutParameter 对象
        """
        return self.create_sa_logout_parameter()

    async def is_valid_token(self, token_value: str) -> bool:
        """
        判断一个 token 是否是有效的
        """
        login_id = await self.get_login_id_by_token(token_value)
        return bool(login_id)

    # ------------------- 注销相关操作 -------------------
    def kickout_by_token_value(
        self, token_value: str, logout_parameter: Optional[LogoutParameter] = None
    ):
        """
        将指定 token 踢下线
        """
        if logout_parameter is None:
            logout_parameter = self.create_sa_logout_parameter()
        logout_parameter.mode = LogoutMode.KICKOUT
        self._logout_by_token_value(token_value, logout_parameter)

    def replaced_by_token_value(
        self, token_value: str, logout_parameter: LogoutParameter
    ):
        """
        将指定 token 顶替下线
        """
        logout_parameter = logout_parameter or self.create_sa_logout_parameter()
        logout_parameter.mode = LogoutMode.REPLACED
        self._logout_by_token_value(token_value, logout_parameter)

    def replaced_by_logout_parameter(self, login_id, param):
        """
        根据登出参数顶替指定账号登录
        """
        logout_parameter = self.create_logout_parameter()
        logout_parameter.mode = LogoutMode.REPLACED
        self._logout(login_id, logout_parameter)

    # ------------------- 会话查询 -------------------

    # ------------------- Session 相关 -------------------
    def get_session(self, is_create: Optional[bool] = None):
        """
        获取 Session 对象
        """
        # 这里应该根据具体需求实现获取 Session 的逻辑
        if is_create is None:
            is_create = True
        login_id = self.get_login_id()
        return self.get_session_by_login_id(login_id, is_create)

    def get_anon_token_session(self):
        """
        获取匿名 Token Session
        """
        # 获取匿名 Token Session 的实现
        pass

    async def get_token_session_by_token(
        self, token_value: str, is_create: bool = False
    ):
        """
        获取指定 token 的 Token-Session
        """
        if not token_value:
            raise TokenError(
                error_type=TokenErrorType.NOT_TOKEN,
                context=dict(
                    login_type=self._login_type,
                    token_value="",
                ),
            )

        # 如果能查询到旧记录，则直接返回
        session_id = self._splicing_key_token_session(token_value)
        token_session = await self.dao.get_session(session_id)
        if token_session:
            return token_session

        # 指定了不需要创建，返回 None
        if not is_create:
            return None

        # 检查一下这个 token 是否为有效 token，无效 token 不允许创建
        if self.config.token_session_check_login and not self.is_valid_token(
            token_value
        ):
            raise TokenError(
                error_type=TokenErrorType.INVALID_TOKEN,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

        # 创建 Token-Session 并返回
        return await self.get_session_by_session_id(
            session_id,
            True,
            None,
            lambda session: self._init_token_session(session, token_value),
        )

    async def get_token_session(self, is_create: bool = True):
        """
        获取当前 token 的 Token-Session
        """
        token_value = self.get_token_value()
        await self.check_active_timeout_by_config(token_value)
        return await self.get_token_session_by_token(token_value, is_create)

    async def get_session_by_login_id(
        self, login_id: Any, is_create: bool = False, timeout: Optional[int] = None
    ) -> AuthSessionModel:
        """
        获取指定账号 id 的 Account-Session
        """
        if not login_id:
            raise Exception("Account-Session 获取失败：loginId 不能为空")

        return await self.get_session_by_session_id(
            self._splicing_key_session(login_id),
            is_create,
            timeout,
            lambda session: self._init_account_session(session, login_id),
        )

    def _init_account_session(self, session: AuthSessionModel, login_id: Any):
        """
        初始化 Account-Session
        """
        session.set_type("account")
        session.set_login_type(self._login_type)
        session.set_login_id(login_id)

    async def get_session_by_session_id(
        self,
        session_id: str,
        is_create: bool = False,
        timeout: Optional[int] = None,
        append_operation: Callable[[AuthSessionModel], None] = None,
    ):
        """
        获取指定 key 的 SaSessionModel
        """
        if not session_id:
            raise Exception("SessionId 不能为空")

        # 先检查这个 SaSessionModel 是否已经存在
        session = await self.dao.get_session(session_id)

        if not session and is_create:
            # 创建这个 SaSessionModel
            session = self._create_session(session_id)

            # 追加操作
            if append_operation:
                append_operation(session)

            # 如果未提供 timeout，则根据相应规则设定默认的 timeout
            if timeout is None:
                if session.session_type == "token":
                    timeout = await self.get_token_timeout(session.token)
                    if timeout == -2:
                        timeout = self.config.timeout
                else:
                    timeout = self.config.timeout

            # 将这个 SaSessionModel 入库
            await self.dao.set_session(session, timeout)

        return session

    def _init_token_session(self, session: AuthSessionModel, token_value: str):
        """
        初始化 Token-Session
        """
        session.set_type("token")
        session.set_login_type(self.login_type)
        session.set_token(token_value)

    async def delete_token_session(self, token_value: str):
        """
        删除指定 token 的 Token-Session
        """
        await self.dao.delete(self._splicing_key_token_session(token_value))

    # ------------------- 活跃度与过期时间 -------------------
    async def update_last_active_to_now(self, token_value: Optional[str] = None):
        """
        续签指定 token：将这个 token 的最后活跃时间更新为当前时间戳
        """
        if token_value is None:
            token_value = self.get_token_value()
        key = self._splicing_key_last_active_time(token_value)
        value = f"{int(time.time() * 1000)},{self.get_token_use_active_timeout(token_value)}"
        await self.dao.update(key, value)

    def check_active_timeout(self, token_value: Optional[str] = None):
        """
        检查指定 token 是否已被冻结，如果是则抛出异常
        """
        if token_value is None:
            token_value = self.get_token_value()
        if self.is_freeze(token_value):
            raise TokenError(
                error_type=TokenErrorType.TOKEN_FREEZE,
                context=dict(
                    login_type=self._login_type,
                    token_value=token_value,
                ),
            )

    async def get_token_use_active_timeout(self, token_value: str) -> Optional[int]:
        """
        获取指定 token 在缓存中的 activeTimeout 值
        """
        # 在未启用动态 activeTimeout 功能时，直接返回 None
        if not self.config.dynamic_active_timeout:
            return None

        # 先取出这个 token 的最后活跃时间值
        key = self._splicing_key_last_active_time(token_value)
        value = await self.dao.get(key)

        # 解析，无值的情况下返回 None
        if value and "," in value:
            return int(value.split(",")[1])  # noqa
        return None

    # ------------------- 角色认证 -------------------
    async def has_role(self, *role_array: str):
        """
        【内部方法】当前账号是否拥有指定角色

        Args:
            *role_array: 一个或多个角色字符串
            logic: 逻辑运算符，"and"表示需要满足所有条件，"or"表示满足任一条件

        Returns:
            bool: 是否满足角色要求
        """
        if not role_array:
            return False
        login_id = await self.get_login_id_default_null()
        if login_id is None:
            return False
        role_list = await self.get_role_list(login_id)
        if not role_list:
            return True
        if "ROLE_ADMIN" in role_list:
            return True
        return self.has_element(role_list, role_array, logic="or")

    def check_role(self, *role_array: str):
        """
        检查当前账号是否同时拥有所有指定角色，不满足则抛出异常
        """
        if not self.has_role(*role_array):
            raise RoleException(login_type=self._login_type, required=role_array)

    # ------------------- 权限认证 -------------------
    async def has_permission(
        self, *permission_array: str, logic: Literal["and", "or"] = "and"
    ):
        """
        【内部方法】当前账号是否拥有指定权限

        Args:
            *permission_array: 一个或多个权限字符串
            logic: 逻辑运算符，"and"表示需要满足所有条件，"or"表示满足任一条件

        Returns:
            bool: 是否满足权限要求
        """
        # 快速检查：如果没有指定权限，则返回False
        if not permission_array:
            return False

        # 快速检查：如果用户未登录，则返回False
        login_id = await self.get_login_id_default_null()
        if login_id is None:
            return False

        # 获取权限列表并进行检查
        permission_list = await self.get_permission_list(login_id)

        # 如果权限列表为空，返回True（表示不需要权限检查）
        if not permission_list:
            return True

        # 使用优化后的权限匹配方法
        return self.has_element(permission_list, permission_array, logic=logic)

    def check_permission(
        self, *permission_array: str, logic: Literal["and", "or"] = "and"
    ):
        """
        检查当前账号是否同时拥有所有指定权限，不满足则抛出异常

        Args:
            *permission_array: 一个或多个权限字符串
            logic: 逻辑运算符，"and"表示需要满足所有条件，"or"表示满足任一条件

        Raises:
            PermissionException: 当权限不足时抛出异常
        """
        if not self.has_permission(*permission_array, logic=logic):
            # 抛出更详细的权限异常信息
            raise PermissionException(
                login_type=self._login_type, required=permission_array
            )

    # ------------------- Token 反查 -------------------

    async def get_terminal_list_by_login_id(
        self, login_id: Any, device_type: Optional[str] = None
    ):
        """
        获取指定账号 id 的终端设备列表
        """
        session = await self.get_session_by_login_id(login_id, False)
        if session is None:
            return []
        return session.get_terminal_list_by_device_type(device_type)

    async def get_terminal_info(self):
        """
        获取当前会话的终端信息
        """
        token_value = self.get_token_value()
        return await self.get_terminal_info_by_token(token_value)

    async def get_login_device_type_by_token(self, token_value: str):
        """
        根据 token 获取登录设备类型
        """
        terminal_info = await self.get_terminal_info_by_token(token_value)
        return terminal_info.device_type if terminal_info else None

    async def get_login_device_id_by_token(self, token_value: str):
        """
        根据 token 获取登录设备 ID
        """
        terminal_info = await self.get_terminal_info_by_token(token_value)
        return terminal_info.device_id if terminal_info else None

    def get_login_device_id(self):
        """
        获取当前会话的登录设备 ID
        """
        token_value = self.get_token_value()
        return self.get_login_device_id_by_token(token_value)

    async def is_trust_device_id(self, login_id: Any, device_id: str):
        """
        判断指定账号 id 是否信任指定设备 ID
        """
        terminal_list = await self.get_terminal_list_by_login_id(login_id)
        return any(terminal.device_id == device_id for terminal in terminal_list)

    # ------------------- 会话管理 -------------------

    async def search_token_value(
        self, keyword: Optional[str], start: int, size: int, sort_type: bool
    ):
        """
        搜索 Token 值
        """
        # 使用 DAO 的 search_data 方法搜索 token 值
        # 使用一个示例token值来获取正确的key前缀
        prefix = self._splicing_key_token_value("")
        keyword = "" if keyword is None else keyword
        return await self.dao.search_data(prefix, keyword, start, size, sort_type)

    async def search_session_id(
        self, keyword: Optional[str], start: int, size: int, sort_type: bool
    ):
        """
        搜索 Session ID
        """
        # 使用 DAO 的 search_data 方法搜索 session ID
        # 使用一个示例login_id来获取正确的key前缀
        prefix = self._splicing_key_session("")
        keyword = "" if keyword is None else keyword
        return await self.dao.search_data(prefix, keyword, start, size, sort_type)

    def search_token_session_id(
        self, keyword: Optional[str], start: int, size: int, sort_type: bool
    ):
        """
        搜索 Token Session ID
        """
        # 使用 DAO 的 search_data 方法搜索 token session ID
        # 使用一个示例token_value来获取正确的key前缀
        prefix = self._splicing_key_token_session("")
        keyword = "" if keyword is None else keyword
        return self.dao.search_data(prefix, keyword, start, size, sort_type)

    async def logout_by_max_login_count(
        self,
        login_id: Any,
        session: Optional[AuthSessionModel],
        device_type: Optional[str],
        max_login_count: int,
        overflow_logout_mode: LogoutMode,
    ):
        """
        根据最大登录次数进行登出处理
        """
        if session is None:
            session = self.get_session_by_login_id(login_id, False)
            if session is None:
                return

        terminal_list = session.get_terminal_list_by_device_type(device_type)

        # 计算需要移除的终端数量
        remove_count = len(terminal_list) - max_login_count
        if remove_count > 0:
            # 移除超出最大登录次数的终端
            for i in range(remove_count):  # noqa
                logout_param = self.create_logout_parameter()
                logout_param.mode = overflow_logout_mode
                await self._remove_terminal(session, terminal_list[i], logout_param)

        await session.logout_by_terminal_count_to_zero()

    async def _logout(self, login_id, logout_parameter):
        # 1、获取此账号的 Account-Session，上面记录了此账号的所有登录客户端数据
        session = await self.get_session_by_login_id(login_id, False)
        if session is not None:
            # 2、遍历此 SaTerminalInfoModel 客户端列表，清除相关数据
            terminal_list = session.terminal_list_copy()
            for terminal in terminal_list:
                # 不符合 deviceType 的跳过
                if (
                    logout_parameter.device_type
                    and logout_parameter.device_type != terminal.device_type
                ):
                    continue
                # 不符合 deviceId 的跳过
                if (
                    logout_parameter.device_id
                    and logout_parameter.device_id != terminal.device_id
                ):
                    continue
                await self._remove_terminal(session, terminal, logout_parameter)

            # 3、如果代码走到这里的时候，此账号已经没有客户端在登录了，则直接注销掉这个 Account-Session
            if logout_parameter.mode == LogoutMode.REPLACED:
                # 因为调用顶替下线时，一般都是在新客户端正在登录，所以此种情况不需要清除该账号的 Account-Session
                # 如果清除了 Account-Session，将可能导致 Account-Session 被注销后又立刻创建出来，造成不必要的性能浪费
                pass
            else:
                await session.logout_by_terminal_count_to_zero()

    async def _remove_terminal(
        self,
        session: AuthSessionModel,
        terminal: TerminalInfoModel,
        logout_parameter: LogoutParameter,
    ):
        """
        移除终端设备

        Args:
            session: 会话对象
            terminal: 终端信息对象
            logout_parameter: 登出参数对象
        """
        login_id = session.get_login_id()
        token_value = terminal.get_token_value()

        # 1、从 Account-Session 上清除此设备信息
        await session.remove_terminal(token_value)

        # 2、清除这个 token 的最后活跃时间记录
        if self.is_open_check_active_timeout():
            await self.clear_last_active(token_value)

        # 3、清除这个 token 的 Token-Session 对象
        if not logout_parameter.is_keep_token_session:
            await self.delete_token_session(token_value)

        # 4、清理或更改 Token 映射
        # 5、发布事件通知

        # SaLogoutMode.LOGOUT：注销下线
        if logout_parameter.mode == LogoutMode.LOGOUT:
            await self.delete_token_to_id_mapping(token_value)
            # 使用统一的事件发布方法
            await self._publish_logout_event("LOGOUT", login_id, token_value)

        # SaLogoutMode.KICKOUT：踢人下线
        if logout_parameter.mode == LogoutMode.KICKOUT:
            await self.update_token_to_id_mapping(token_value, LoginIdMark.KICK_OUT)
            # 使用统一的事件发布方法
            await self._publish_logout_event("KICKOUT", login_id, token_value)

        # SaLogoutMode.REPLACED：顶人下线
        if logout_parameter.mode == LogoutMode.REPLACED:
            await self.update_token_to_id_mapping(token_value, LoginIdMark.BE_REPLACED)
            # 使用统一的事件发布方法
            await self._publish_logout_event("REPLACED", login_id, token_value)

    async def replaced_by_type(self, login_id, device_type):
        """
        根据设备类型顶替指定账号登录
        """
        logout_parameter = self.create_logout_parameter()
        logout_parameter.mode = LogoutMode.REPLACED
        logout_parameter.device_type = device_type
        await self._logout(login_id, logout_parameter)

    def is_open_check_active_timeout(self):
        """
        判断是否开启活跃度检查
        """
        return (
            self.config.active_timeout != AuthTokenDao.NEVER_EXPIRE
            or self.config.dynamic_active_timeout
        )

    async def get_session_timeout(self):
        """
        获取当前会话的剩余有效时间
        """
        login_id = await self.get_login_id_default_null()
        if login_id is None:
            return AuthTokenDao.NOT_VALUE_EXPIRE
        return await self.get_session_timeout_by_login_id(login_id)

    async def get_token_session_timeout(self):
        """
        获取当前 Token 会话的剩余有效时间
        """
        token_value = self.get_token_value()
        if not token_value:
            return -2  # NOT_VALUE_EXPIRE
        return await self.get_token_session_timeout_by_token_value(token_value)

    async def get_token_active_timeout(self) -> int:
        """
        获取当前 Token 的活跃剩余有效时间
        """
        token_value = self.get_token_value()
        if not token_value:
            return -2  # NOT_VALUE_EXPIRE
        return await self.get_token_active_timeout_by_token(token_value)

    async def kickout(
        self,
        login_id: Any,
        device_type: Optional[str] = None,
        logout_parameter: Optional[LogoutParameter] = None,
    ):
        """
        顶替指定账号登录
        """
        if logout_parameter is None:
            logout_parameter = self.create_logout_parameter()
        if device_type:
            logout_parameter.device_type = device_type
        logout_parameter.mode = LogoutMode.KICKOUT
        await self._logout(login_id, logout_parameter)

    async def replaced(
        self,
        login_id: Any,
        device_type: Optional[str] = None,
        logout_parameter: Optional[LogoutParameter] = None,
    ):
        """
        顶替指定账号登录
        """
        if logout_parameter is None:
            logout_parameter = self.create_logout_parameter()
        if device_type:
            logout_parameter.device_type = device_type
        logout_parameter.mode = LogoutMode.REPLACED
        await self._logout(login_id, logout_parameter)

    async def disable(self, login_id: Any, timer: int, service: Optional[str] = None):
        """

        Args:
            login_id: 账号id
            timer: 禁用时间
            service: 禁用服务

        Returns:

        """
        if not service:
            service = TokenConsts.DEFAULT_DISABLE_SERVICE
        await self.disable_level(
            login_id,
            TokenConsts.DEFAULT_DISABLE_LEVEL,
            timer,
            service,
        )

    async def disable_level(
        self, login_id: Any, level: int, timer: int, service: Optional[str] = None
    ):
        if not login_id:
            raise TokenError(
                error_type=TokenErrorType.TOKEN_FREEZE,
                context=dict(
                    login_type=self._login_type,
                    token_value="",
                ),
            )
        await self.dao.set(
            self._splicing_key_disable(login_id, service), str(level), timer
        )

    async def is_disable(self, login_id: Any, service: Optional[str] = None) -> bool:
        """
        判断一个账号是否被禁用

        Args:
            login_id: 账号ID
            service: 服务标识，默认为None

        Returns:
            bool: 如果账号被禁用返回True，否则返回False
        """
        if not service:
            service = TokenConsts.DEFAULT_DISABLE_SERVICE
        return await self.is_disable_level(
            login_id, service, TokenConsts.MIN_DISABLE_LEVEL
        )

    async def is_disable_level(self, login_id: Any, service: str, level: int) -> bool:
        """
        判断一个账号是否被禁用到指定级别

        Args:
            login_id: 账号ID
            level: 禁用级别
            service: 服务标识，默认为None

        Returns:
            bool: 如果账号被禁用到指定级别或更高级别返回True，否则返回False
        """
        disable_level = await self.get_disable_level(login_id, service)
        if disable_level == TokenConsts.NOT_DISABLE_LEVEL:
            return False
        # // 2、再判断被封禁的等级是否达到了指定级别
        return disable_level >= level

    async def get_disable_level(
        self, login_id: Any, service: Optional[str] = None
    ) -> int:
        """
        获取指定账号的禁用级别

        Args:
            login_id: 账号ID
            service: 服务标识，默认为None

        Returns:
            int: 禁用级别，NOT_DISABLE_LEVEL表示未禁用
        """
        value = await self.dao.get(self._splicing_key_disable(login_id, service))
        if value:
            return int(value)
        return TokenConsts.NOT_DISABLE_LEVEL

    async def check_disable(self, login_id: Any, service: Optional[str] = None):
        """
        校验一个账号是否被禁用，如果被禁用则抛出异常

        Args:
            login_id: 账号ID
            service: 服务标识，默认为None

        Raises:
            Exception: 如果账号被禁用则抛出异常
        """
        if await self.is_disable(login_id, service):
            raise TokenError(
                error_type=TokenErrorType.TOKEN_FREEZE,
                context=dict(
                    login_type=self._login_type,
                    token_value="",
                ),
            )

    async def get_disable_time(
        self, login_id: Any, service: Optional[str] = None
    ) -> int:
        """
        获取指定账号禁用的剩余时间

        Args:
            login_id: 账号ID
            service: 服务标识，默认为None

        Returns:
            int: 剩余禁用时间（秒），-1表示永久禁用，-2表示未禁用
        """
        if not service:
            service = TokenConsts.DEFAULT_DISABLE_SERVICE
        timeout = await self.dao.get_timeout(
            self._splicing_key_disable(login_id, service)
        )
        return timeout

    async def untie_disable(self, login_id: Any, services: Optional[list[str]] = None):
        """
        解除账号禁用

        Args:
            login_id: 账号ID
            services: 服务标识，默认为None
        """
        if login_id is None:
            raise TokenError(
                error_type=TokenErrorType.TOKEN_FREEZE,
                context=dict(
                    login_type=self._login_type,
                    token_value="",
                ),
            )
        if not services:
            services = [TokenConsts.DEFAULT_DISABLE_SERVICE]
        for service in services:
            await self.dao.delete(self._splicing_key_disable(login_id, service))
            # 发布事件

    def create_login_parameter(self) -> LoginParameter:
        """
        创建登录参数对象

        Returns:
            LoginParameter: 登录参数对象
        """

        return LoginParameter.from_config(self.config)

    # ------------------- 注销 (会话管理辅助方法) ----------------------------

    async def remove_terminal_by_kickout(
        self, session: AuthSessionModel, terminal: TerminalInfoModel
    ):
        """
        根据踢出模式，删除指定终端的登录信息
        """
        logout_parameter = self.create_logout_parameter()
        logout_parameter.mode = LogoutMode.KICKOUT
        await self._remove_terminal(session, terminal, logout_parameter)

    async def remove_terminal_by_replaced(
        self, session: AuthSessionModel, terminal: TerminalInfoModel
    ):
        """
        根据顶替模式，删除指定终端的登录信息
        """
        logout_parameter = self.create_logout_parameter()
        logout_parameter.mode = LogoutMode.REPLACED
        await self._remove_terminal(session, terminal, logout_parameter)

    async def remove_terminal_by_logout(
        self, session: AuthSessionModel, terminal: TerminalInfoModel
    ):
        """
        根据登出模式，删除指定终端的登录信息
        """
        logout_parameter = self.create_logout_parameter()
        logout_parameter.mode = LogoutMode.LOGOUT
        await self._remove_terminal(session, terminal, logout_parameter)

    def set_dao(self, dao):
        self.dao = dao
